package com.rock.code.province.tj.busi.socketdr;

import com.rock.code.provider.es.controller.ElasticSearchInterfaceController;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
import java.util.logging.Logger;

/**
 * 消费者 Consumer.
 *
 * @author sunhaobo
 * @date 2019 /09/10
 */
public class Consumer {
    /**
     * The Kafka consumer.
     */
    KafkaConsumer<String, String> kafkaConsumer;
    /**
     * The Logger.
     */
    static Logger logger = Logger.getLogger(Consumer.class.getName());

    /**
     * Instantiates a new Consumer.
     *
     * @author sunhaobo
     * @date 2019 /09/27
     */
    public Consumer() {
        //1.配置属性值
        Properties props = new Properties();
        //kafka是服务器地址
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.43.74:9092");
        //定义消费者组
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "groupid");
        //自动提交（offset）
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        //自动处理的间隔时间1秒
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);
        //key和values的持久化设置
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        //2.创建消费者
        kafkaConsumer = new KafkaConsumer<>(props);
        //重置偏移量
//        TopicPartition partition0 = new TopicPartition("topicName", 0);
//        kafkaConsumer.seekToEnd(Arrays.asList(partition0));
    }

    /**
     * Consumer.
     *
     * @author sunhaobo
     * @date 2019 /09/27
     */
    public void consumer() throws IOException {
        List list = new ArrayList<>();
        //3.订阅消费topic(可以有多个topic)
        kafkaConsumer.subscribe(Arrays.asList("test"));
        //4.执行消费的操作
        while (true) {
            //100ms消费一次
            // kafkaConsumer.poll(100)读出来，读到records
            ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("-----------------");
                //打印偏移量，key，value
                System.out.printf("offset = %d, value = %s", record.offset(), record.key(), record.value());
                System.out.println();
                logger.info(record.toString());
                list.add(record.value());
            }
            //调用es新增方法
            ElasticSearchInterfaceController es = new ElasticSearchInterfaceController();
            es.addKpis("es的索引名称", list);
        }
    }

    /**
     * Main.
     *
     * @param args the args
     * @author sunhaobo
     * @date 2019 /09/10
     */
    public static void main(String[] args) {

    }

}
